
namespace Stee.CAP8.MessageBus.Kafka.Admin
{
    using System;
    using System.Collections.Generic;
    using System.Threading.Tasks;
    using Confluent.Kafka;
    using Confluent.Kafka.Admin;
    using NLog;
    using Stee.CAP8.MessageBus.Interface.Models;

    public class AdminClientImpl
    {
        #region Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        #endregion Static Variables


        #region Fields

        /// <summary>
        /// Consumer configuration
        /// </summary>
        private readonly AdminClientConfig adminClientConfig;

        /// <summary>
        /// Kafka consumer builder
        /// </summary>
        private AdminClientBuilder adminClientBuilder;

        /// <summary>
        /// Kafka consumer
        /// </summary>
        private IAdminClient adminClient;

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets an optional identifier of a Kafka consumer(in a consumer group) that is passed to a Kafka broker with every request.
        /// The sole purpose of this is to be able to track the source of requests beyond just ip and port
        /// by allowing a logical application name to be included in Kafka logs and monitoring aggregates.
        /// </summary>
        public string ClientID { get; private set; } = string.Empty;

        #endregion Properties


        #region Constructor

        /// <summary>
        /// Initializes a new instance of the AdminClientImpl class
        /// </summary>
        /// <param name="connectionProperties"></param>
        public AdminClientImpl(IDictionary<string, string> connectionProperties)
        {
            //// Create configuration
            //// Use connection properties here
            //// https://docs.confluent.io/current/installation/configuration/admin-configs.html
            //// TODO: Build configuration reader
            this.adminClientConfig = new AdminClientConfig
            {
                BootstrapServers = "127.0.0.1:9092", // Mandatory config
                ClientId = this.ClientID,
            };
        }

        #endregion Constructor


        #region Interface Implementation

        public async Task CreateTopicsAsync(TopicConfiguration topicConfiguration)
        {
            try
            {
                //// Initialize builder
                this.InitializeProducerBuilder();

                using (this.adminClient = this.adminClientBuilder.Build())
                {
                    try
                    {
                        TopicSpecification topicSpec = new TopicSpecification
                        {
                            Name = topicConfiguration.Name,
                            NumPartitions = topicConfiguration.NumOfPartitions,
                            ReplicationFactor = topicConfiguration.ReplicationFactor,
                        };

                        ////// Create options
                        //CreateTopicsOptions topicOptions = new CreateTopicsOptions();
                        //topicOptions.ValidateOnly = false;

                        //// Create topic
                        await this.adminClient.CreateTopicsAsync(new TopicSpecification[] { topicSpec }, null).ConfigureAwait(false);
                    }
                    catch (CreateTopicsException e)
                    {
                        Console.WriteLine($"An error occured creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
                    }
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
            finally
            {
                this.DisposeAdminClient();
            }
        }


        public void CreatePartitionsAsync()
        {
            try
            {

            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }


        public void DeleteTopicsAsync()
        {
            try
            {

            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }


        public void DescribeConfigsAsync()
        {
            try
            {

            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        #endregion Interface Implementation


        #region Private Methods

        /// <summary>
        /// Initializes the producer builder
        /// </summary>
        private void InitializeProducerBuilder()
        {
            this.adminClientBuilder = new AdminClientBuilder(this.adminClientConfig);
            //this.adminClientBuilder.SetLogHandler();
            //this.adminClientBuilder.SetErrorHandler();
            //this.adminClientBuilder.SetStatisticsHandler(Serializers.ByteArray);
        }

        /// <summary>
        /// Dispose admin client
        /// </summary>
        private void DisposeAdminClient()
        {
            if (this.adminClient != null)
            {
                this.adminClient.Dispose();
                this.adminClient = null;
            }
        }

        #endregion Private Methods
    }
}
